#include "ThreadsPool.h"

ThreadsPool *pool;

ThreadsPool::ThreadsPool(size_t workerThreads_) :
	workerThreads(workerThreads_),
	threads(new pthread_t[workerThreads]),
	jobQueues(new JobsQueue[workerThreads + 1]),
	lastQueue(0)
{
/*
	for (size_t i = 0; i < workerThreads + 1; i++)
		jobQueues[i] = JobsQueue();
*/
	initializeThreads();
}

ThreadsPool::~ThreadsPool()
{
	delete[] jobQueues;
	delete[] threads;
}

#ifdef	DEBUG
#include <iostream>
using namespace std;
#endif	/* DEBUG */

/* A static function that is activated by the pthread_create function.
   Receives a pointer to ThreadData as its argument */
static void *thread_wrapper(void *args)
{
	ThreadsPool::ThreadData *td = (ThreadsPool::ThreadData *)args;
#ifdef	DEBUG
	cout << "Worker thread " << td->id << " starts execution" << endl;
#endif	/* DEBUG */
	/* Execute jobs until the end... */
	while (true)
	{
		/* TODO - if we want to add stealing, the waiting should be timed out, and then trying to fetch from another queue */
		/* Wait for a job in the queue */
		Job *curJob = td->queue.waitForJob();

		if (curJob == NULL)
		{
#ifdef	DEBUG
			cout << "Worker thread " << td->id << " stops execution" << endl;
#endif	/* DEBUG */
			delete td;
			return NULL;
		}

#ifdef	DEBUG
		cout << "Worker thread " << td->id << " processes a job of type = " << curJob->getType() << endl;
#endif	/* DEBUG */
		/* Process the job */
		curJob->process();
#ifdef	DEBUG
		cout << "Worker thread " << td->id << " finished processing job" << endl;
#endif	/* DEBUG */
		(*myDoneJobs)[td->id].push_back(curJob);
	}
}

void ThreadsPool::initializeThreads()
{
#ifdef	DEBUG
	cout << "Initializing " << workerThreads << " threads" << endl;
#endif	/* DEBUG */
	for (size_t i = 0; i < workerThreads; i++)
	{
		ThreadData *td = new ThreadData(i, jobQueues[i]);
		pthread_create(&threads[i], NULL, &thread_wrapper, td);
	}
}

void ThreadsPool::addJob(Job *job)
{
	/* Not perfect - might cause synchronization issues */
	size_t curQueue = __sync_fetch_and_add( &lastQueue, 3);
	jobQueues[curQueue % (workerThreads + 1)].addJob(job);
	//lastQueue = (lastQueue + 1) % (workerThreads + 1);
}

void ThreadsPool::noMoreJobs()
{
#ifdef	DEBUG
	cout << "ThreadsPool sending No-More-Jobs to threads" << endl;
#endif	/* DEBUG */
	/* Insert a dummy job to all queues */
	for (size_t i = 0; i < workerThreads + 1; i++)
	{
		jobQueues[i].addJob(NULL);
	}
}

void ThreadsPool::stopThreads()
{
#ifdef	DEBUG
	cout << "ThreadsPool stopping threads" << endl;
#endif	/* DEBUG */
	/* Wait for the worker threads */
	for (size_t i = 0; i < workerThreads; i++)
	{
		pthread_join(threads[i], NULL);
	}
}

JobsQueue& ThreadsPool::getMainThreadQueue()
{
	return jobQueues[workerThreads];
}

